package cn.wisenergy.pai.monitor.worker;

import java.io.IOException;
import java.util.Properties;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import cn.wisenergy.pai.monitor.Const;
import cn.wisenergy.pai.monitor.linux.parser.LinuxSystemTool;
import cn.wisenergy.pai.monitor.linux.parser.TimeNode;
import cn.wisenergy.pai.monitor.utils.ConfigurationTools;

public class MessageSender {
	
	private static final int sleepTime=5000;

	/**
	 * @param args
	 * @throws IOException 
	 */
	public static void main(String[] args) throws IOException {
		final Properties pps = new Properties();
		final String masterIp=ConfigurationTools.getPAIConfValue("JOB_SERVER_ADDR");
		final String broker=masterIp+":9092";
		pps.put("metadata.broker.list",broker);
		pps.put("request.required.acks", "1");
		final Producer<byte[], byte[]> producer = new Producer<byte[], byte[]>(new ProducerConfig(pps));
		try {
			while(true){
				 try {   
		    			TimeNode t=LinuxSystemTool.getCurrentTimeNode();
		    			String key=t.getHostspot().getIp()+"_"+t.getTime().getTime();
		    			KeyedMessage<byte[],byte[]> message=new KeyedMessage<byte[],byte[]>(Const.TOPIC, key.getBytes(), t.toBytes());
		    			producer.send(message);
		            } catch (Exception e) {   
		                e.printStackTrace();   
		            }
				 Thread.sleep(sleepTime); 
			}
		} catch (Exception e1) {
			e1.printStackTrace();
		}
		producer.close();
	}

}
